﻿using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsumerTest1
{
    /// <summary>
    /// 这个就相当于是java的那个“独立消费者”，即消费都固定监听指定分区，这样就不会触发再平衡
    /// </summary>
    public class ManualAssignConsumer : IConsumer
    {
        public Task ExcuteAsync(CancellationToken token = default)
        {
            ConsumerConfig config = new ConsumerConfig();
            config.BootstrapServers = "192.168.3.68:9092,192.168.3.66:9092,192.168.3.69:9092";
            config.GroupId = "static-consumer-test";
            config.EnableAutoCommit = true;
            using (IConsumer<string, string> consumer = new ConsumerBuilder<string, string>(config)
                .SetErrorHandler((_, e) => Console.WriteLine($"Error:{e.Reason}"))
                .Build())
            {
                //从指定主题指定分区的最开始位置读起
                consumer.Assign(new TopicPartitionOffset("staticconsumer", 0, Offset.Beginning));
                try
                {
                    while (true)
                    {
                        try
                        {
                            var consumerResult = consumer.Consume(token);
                            Console.WriteLine($"Received message at {consumerResult.TopicPartitionOffset}: {consumerResult.Message.Value}");
                        }
                        catch (ConsumeException e)
                        {
                            Console.WriteLine($"Consume error: {e.Error.Reason}");
                        }
                    }
                }
                catch (OperationCanceledException e)
                {
                    Console.WriteLine("Closing consumer.");
                    consumer.Close();
                }
            }
            return Task.CompletedTask;
        }
    }
}
